package me.shiqiang.im.listeners;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.tst.common.utils.JSONUtils;
import me.shiqiang.im.enums.MessageCmd;
import me.shiqiang.im.msg.DistributeMsgBody;
import me.shiqiang.im.msg.MsgBody;
import me.shiqiang.im.msg.NewMsg;
import me.shiqiang.im.redis.RedisKey;
import me.shiqiang.im.service.ChatSwitchService;
import me.shiqiang.im.service.MsgService;
import me.shiqiang.im.service.ServerNodeClient;
import me.shiqiang.im.utils.JsonUtils;
import me.shiqiang.im.util.RedisPipelineProcessor;
import io.socket.client.Socket;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Created by maliangxue on 2018/4/15.
 */
@Component("GroupMsgListener")
public class GroupMsgListener extends RouterServerListener  {
    private Logger logger = LoggerFactory.getLogger(this.getClass());


    @Value("${send.topic}")
    private String topic;

    @Autowired
    MsgService service;

    @Value("${server.id}")
    String serverId;

    @Autowired
    ServerNodeClient serverNodeClient;

    @Autowired
    ChatSwitchService chatSwitchService;

    @Autowired
    Socket socket;

    @Autowired
    ConcurrentLinkedQueue<RedisPipelineProcessor> queue;

    @Autowired
    StringRedisTemplate redisTemplate;


    @Override
    public String eventName() {
        return "groupMsg";
    }

    @Override
    public void call(Object... objects) {
        JSONObject obj = (JSONObject)objects[0];
        String msgBodyStr = obj.toString();
        try {
            logger.info("群收到消息: {}", msgBodyStr);
            DistributeMsgBody msgBody = JsonUtils.getInstance().fromJson(msgBodyStr, DistributeMsgBody.class);
            Long groupId = msgBody.getTo();
            Long syncKey = msgBody.getSyncKey();
            Set<String> userIds = redisTemplate.opsForSet().members(RedisKey.groupUserSet(groupId));
            Set<String> atUserIds = new HashSet<>();
            boolean atAll = false;
            MsgBody toClientMsgBody = new MsgBody().from(msgBody);

            for (Long at : msgBody.getAt()) {
                if (at == -1) { // -1代表所有人
                    atAll = true;
                    break;
                }
                atUserIds.add(String.valueOf(at));
            }
            boolean finalAtAll = atAll;
            queue.add(connection -> {
                try {
                    if (finalAtAll) {
                        toClientMsgBody.setAt(1);
                    }
                    for (String userId : userIds) {
                        if (!chatSwitchService.chatSwitch(Long.valueOf(userId))) {
                            continue;
                        }
                        if (!atUserIds.isEmpty()) {
                            if (atUserIds.contains(userId)) {
                                toClientMsgBody.setAt(1);
                            }else {
                                toClientMsgBody.setAt(0);
                            }
                        }

                        NewMsg fromNewMsg = new NewMsg(Long.parseLong(userId), syncKey, toClientMsgBody);
                        if (redisTemplate.opsForZSet().zCard(RedisKey.clientMsgQueue(userId)) == null || redisTemplate.opsForZSet().zCard(RedisKey.clientMsgQueue(userId)) == 0) {
                            connection.zAdd(RedisKey.clientMsgQueue(userId).getBytes(), syncKey, JSONUtils.toJSON(toClientMsgBody).getBytes());
                            connection.expire(RedisKey.clientMsgQueue(userId).getBytes(), 604800);//7天
                        }else {
                            connection.zAdd(RedisKey.clientMsgQueue(userId).getBytes(), syncKey, JSONUtils.toJSON(toClientMsgBody).getBytes());
                        }
                        socket.emit(MessageCmd.NEW_MSG, new JSONObject(JsonUtils.getInstance().toJson(fromNewMsg)));
                    }
                } catch (JsonProcessingException e) {
                    logger.error("转换异常", e);
                }
            });
        } catch (Exception e) {
            logger.error(String.format("消息处理失败：%s", msgBodyStr), e);
        }
    }
}
